package com.hsogoo.avatar.transport.netty;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.ThreadFactory;

import com.hsogoo.avatar.constants.AvatarConstant;
import com.hsogoo.avatar.enums.Protocol;
import com.hsogoo.avatar.exceptions.AvatarException;
import com.hsogoo.avatar.model.AvatarAddress;
import com.hsogoo.avatar.transport.AvatarChannel;
import com.hsogoo.avatar.transport.AvatarConnection;
import com.hsogoo.avatar.transport.AvatarConnectionManager;
import com.hsogoo.avatar.transport.AvatarConnector;
import com.hsogoo.avatar.transport.AvatarProcessor;
import com.hsogoo.avatar.transport.netty.handler.AvatarConnectorHandler;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.DefaultThreadFactory;

/**
 * @Author huangsaigang
 * Created @2018/9/29.
 */
public class AvatarNettyConnector implements AvatarConnector {

    protected final Protocol protocol;
    private Bootstrap bootstrap;
    private EventLoopGroup worker;
    private AvatarConnectionManager connectionManager = new AvatarConnectionManager();
    private int nWorkers;
    private AvatarProcessor processor;

    public AvatarNettyConnector(Protocol protocol) {
        this(protocol, AvatarConstant.AVAILABLE_PROCESSORS << 1);
    }

    public AvatarNettyConnector(Protocol protocol, int nWorkers) {
        this.protocol = protocol;
        this.nWorkers = nWorkers;
        init();
    }

    protected void init() {
        ThreadFactory workerFactory = workerThreadFactory("avatar.connector");
        worker = initEventLoopGroup(nWorkers, workerFactory);
        bootstrap = new Bootstrap().group(worker);
        setOptions();
        bootstrap.channelFactory(SocketChannelProvider.AVATAR_CONNECTOR);
    }

    private void setOptions(){

    }

    private EventLoopGroup initEventLoopGroup(int nThreads, ThreadFactory tFactory) {
        return new NioEventLoopGroup(nThreads, tFactory);
    }

    private ThreadFactory workerThreadFactory(String name) {
        return new DefaultThreadFactory(name, Thread.MAX_PRIORITY);
    }

    @Override
    public AvatarConnection connect(AvatarAddress address) {

        ChannelFuture future;
        final SocketAddress socketAddress = InetSocketAddress.createUnresolved(address.getHost(), address.getPort());
        try {
            bootstrap.handler(new ChannelInitializer<Channel>() {

                @Override
                protected void initChannel(Channel ch) throws Exception {
                    ch.pipeline().addLast(new ProtocolDecoder());
                    ch.pipeline().addLast(new AvatarConnectorHandler(processor));
                    ch.pipeline().addLast(new ProtocolEncoder());
                }
            });
            future = bootstrap.connect(socketAddress).sync();
        } catch (Throwable t) {
            throw new AvatarException("Connects to [" + address + "] fails", t);
        }
        AvatarChannel nettyChannel = new AvatarNettyChannel(future.channel(),connectionManager, worker);
        return new AvatarNettyConnection(address, nettyChannel);
    }

    public AvatarConnectionManager connectionManager(){
        return connectionManager;
    }

    public AvatarProcessor processor() {
        return processor;
    }

    public AvatarConnector withProcessor(AvatarProcessor processor){
        this.processor = processor;
        return this;
    }

    @Override
    public Protocol protocol() {
        return protocol;
    }
    
}
